package com.open.ssm.mqtt.sub;

import java.util.ArrayList;

import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;

/**
 * MQTT订阅，MqttClient订阅主题
 *
 * @version: v1.0.0
 * @author: wangpf
 * @date: 2021年11月11日 下午6:51:49
 */
public class SubFromMqtt {
	
	protected static Logger logger = LoggerFactory.getLogger(SubFromMqtt.class);

    //订阅多个主题 主题数组  
	protected MqttClient mqttClient;
	protected String clientId = "";
	protected ArrayList<String> topicList = Lists.newArrayList();
    protected String serverURI = "";
	protected String username = "";
	protected String password = "";
	protected int connectionTimeout = 30;
	protected int keepAliveInterval = 60;
	
	public MqttCallback mqttCallback;
	
	/**
	 * 启动订阅者
	 */
	public void startSubscribe() {
		try {
			// clientId为连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientId的保存形式，默认为以内存保存
			mqttClient = new MqttClient(serverURI, clientId, mqttClientPersistence());
			// MQTT的连接设置
			MqttConnectOptions options = new MqttConnectOptions();
			// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
			options.setCleanSession(true);
			// 设置连接的用户名
			options.setUserName(username);
			// 设置连接的密码
			options.setPassword(password.toCharArray());
			// 设置超时时间 单位为秒
			options.setConnectionTimeout(connectionTimeout);
			// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
			options.setKeepAliveInterval(keepAliveInterval);
			// 设置回调
			mqttClient.setCallback(mqttCallback);
			mqttClient.connect(options);
			
			//订阅多个主题数组
		    String[] TOPICS = new String[topicList.size()];
            String[] topicGroup = topicList.toArray(TOPICS);
            
            //每一个订阅主题的 QOS=1
            int[] qos = new int[topicList.size()];
            for(int i=0;i<topicList.size();i++) {
            	qos[i]=1;
            }
            
			mqttClient.subscribe(topicGroup, qos);
			logger.info("SubFromMQTT-->订阅成功："+topicList.toString());
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}
	
	// 创建MqttClientPersistence对象
	private MqttClientPersistence mqttClientPersistence() {
        return new MemoryPersistence();
    }

	/**
	 * 订阅者健康检测
	 */
	public boolean isConnected() {
		boolean ret = false;
		if (mqttClient != null 
				&& mqttClient.isConnected()) {
			ret = true;
		}
		return ret;
	}

	/**
	 * 关闭订阅
	 */
	public void close() {
		if (mqttClient != null) {
			try {
				//mqttClient.disconnect();
				mqttClient.close();
				mqttClient = null;
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}
	}
	
	/**
	 * 断开并关闭订阅
	 */
	public void disconnect() {
		if (mqttClient != null) {
			try {
				mqttClient.disconnect();
				mqttClient.close();
				mqttClient = null;
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}
	}

	public MqttClient getMqttClient() {
		return mqttClient;
	}
}
